package salt4j.tm;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 *  An upgradeable, deadlock-detecting, read-write lock.
 */
public class TmLock {

    //list of deferred changes to make when next we DEADLOCK.
    private static final ThreadLocal<ArrayList<Runnable>> TO_APPLY = 
    new ThreadLocal<ArrayList<Runnable>>(){
        protected ArrayList<Runnable> initialValue() { return new ArrayList<Runnable>(); }
    };
    
    /** The MAIN class start here */
    final ReentrantReadWriteLock rw = new ReentrantReadWriteLock(true);
    int writeLockAcquisitions = 0;

    private final HashSet<Thread> owners = new HashSet<Thread>();
    
    void tryLock(final Lock l, boolean isWriteLock) throws LockException {
        boolean waitForSet = false;
        final long giveUpTime = System.currentTimeMillis() + 10000; //this magic number is ugly
        final Thread thisThread = Thread.currentThread();
        for(int i = 0; System.currentTimeMillis() < giveUpTime; i++) {
            boolean success = false;
            success = l.tryLock();
            if (success) {
                if (isWriteLock) writeLockAcquisitions ++; // need not be an atomicinteger (wlock guarantee)
                register();
                
                // unnecessary synchronization eliminated; 25% speedup observed.
                if (waitForSet) {
                    synchronized(DEADLOCK) {
                        WAITFOR.remove(thisThread); owners.add(thisThread);
                    }
                } else {
                    TO_APPLY.get().add(new Runnable() {
                        public void run() { owners.add(thisThread); }
                    });
                }
                return;
            } else {
                boolean deadlocked;

                synchronized(DEADLOCK) {
                    //apply any pending changes to owner now:
                    final ArrayList<Runnable> toApply = TO_APPLY.get();
                    if (!toApply.isEmpty()) { for (Runnable r: toApply){r.run();} toApply.clear(); }
                    if (!waitForSet) { WAITFOR.put(thisThread, this); waitForSet = true; }
                    deadlocked = deadLocked();
                }
                
                if (deadlocked) break; else {
                    synchronized(this) {
                        try {this.wait();}catch(InterruptedException e){}
                    }
                }
            }
        }
        if (waitForSet) synchronized(DEADLOCK) { WAITFOR.remove(thisThread); }
        throw new LockException();
    }

    private static final Object DEADLOCK = new Object();

    boolean deadLocked() { return deadLocked(new HashSet<Thread>(), Thread.currentThread()); }
    private boolean deadLocked(HashSet<Thread> alreadyChecked, Thread currentThread) {
        synchronized(DEADLOCK) { //stop the world
            //WAITFOR.put(currentThread, this);
            for(Thread thread: owners) {
                if (thread == currentThread) return true;
                else if (!alreadyChecked.contains(thread)) {
                    alreadyChecked.add(thread);
                    TmLock tm = WAITFOR.get(thread);
                    if (tm != null && tm.deadLocked(alreadyChecked, currentThread)) return true;
                }
            }
            //WAITFOR.remove(currentThread);
            return false;
        }
    }

    public final void unlockFully() {
        final int nWriteLocks = rw.getWriteHoldCount();
        if (nWriteLocks > 0) if (nWriteLocks == 1) rw.writeLock().unlock(); else throw new Error();
        final int nReadLocks = rw.getReadHoldCount();
        if (nReadLocks > 0) if (nReadLocks == 1) rw.readLock().unlock(); else throw new Error();
        synchronized(DEADLOCK) { owners.remove(Thread.currentThread()); }
        synchronized(this) { this.notifyAll(); }
    }

    private final void upgradeLock() {
        final int initialWrites = writeLockAcquisitions;
        unlockFully(); //drop read lock.
        tryLock(rw.writeLock(), true); //try to acquire write lock immediately.
        if (writeLockAcquisitions != initialWrites + 1) throw new LockException();
    }

    /** Acquire a write lock. May upgrade a read lock. */
    public final void write(){
        if(rw.isWriteLockedByCurrentThread()) return; //only do this once.
        else if (rw.getReadHoldCount() > 0) upgradeLock(); //read to write lock
        else tryLock(rw.writeLock(), true); // acquire the lock.
    }

    /** Acquire a read lock. */
    public final void read() {
        if (rw.isWriteLockedByCurrentThread()) return;  //write lock supercedes read lock
        else if(rw.getReadHoldCount()>0) return; //don't acquire a read lock more than once.
        else tryLock(rw.readLock(), false); //ok, get the lock.
    }

    private void register() { LOCKS.get().add(this); }

    public static class LockException extends RuntimeException{
        public LockException rollback() { return this; }
    }

    private static ThreadLocal<ArrayList<Runnable>> UNDOLOG = new ThreadLocal<ArrayList<Runnable>>(){
        protected ArrayList<Runnable> initialValue() { return new ArrayList<Runnable>(); }
    };

    private static ThreadLocal<Set<TmLock>> LOCKS = new ThreadLocal<Set<TmLock>>(){
        protected Set<TmLock> initialValue() { return new HashSet<TmLock>(); }
    };

    private final static HashMap<Thread, TmLock> WAITFOR = new HashMap<Thread, TmLock>();
 
    //reverse history with undo log:
    private static void rollbackWithUndoLog() {
        ArrayList<Runnable> x = UNDOLOG.get();
        for (int i=x.size()-1; i>=0; i--) x.get(i).run();
        x.clear();
    }

    private static void unlockAll() {
        for (TmLock lock: LOCKS.get()) lock.unlockFully();
        LOCKS.get().clear();
        TO_APPLY.get().clear();
    }
    public static void rollback() { rollbackWithUndoLog(); unlockAll(); }

    public static void commit() { unlockAll(); }

    public static void addToUndoLog(Runnable runnable) { UNDOLOG.get().add(runnable); }
}